// SPDX-License-Identifier: Mulan PSL v2
/*
 * Copyright (c) 2025 Huawei Technologies Co., Ltd.
 * This software is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *         http://license.coscl.org.cn/MulanPSL2
 *
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */

//! New request recorder implementation (based on direct Request serialization)
#![allow(dead_code)]

use std::collections::VecDeque;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Instant, SystemTime, UNIX_EPOCH};

use crate::ipc::bytewise::BytewiseWrite;
use crate::types::xgpu_recorder::v1::*;

use super::{
    format::{ProtobufReader, ProtobufWriter, RecordReader, RecordWriter},
    serializer, RecorderConfig, RecorderError, RecorderStats,
};

/// Get current thread ID (using process PID as alternative implementation)
fn get_current_thread_id() -> u64 {
    use std::thread;
    // Use thread ID hash value as unique identifier
    let thread_id = thread::current().id();
    use std::collections::hash_map::DefaultHasher;
    use std::hash::{Hash, Hasher};
    let mut hasher = DefaultHasher::new();
    thread_id.hash(&mut hasher);
    hasher.finish()
}

/// Serialize BytewiseWrite object to byte array
fn serialize_bytewise<T: BytewiseWrite>(obj: &T) -> Result<(Vec<u8>, u32), RecorderError> {
    // Use new binary serializer
    let serialized_data = serializer::serialize_record(obj)?;

    // Calculate CRC32 checksum
    let crc32_checksum = crc32fast::hash(&serialized_data);

    Ok((serialized_data, crc32_checksum))
}

/// Deserialize BytewiseReadOwned object (for player module)
pub fn deserialize_bytewise<T>(data: &mut [u8], expected_crc32: u32) -> Result<T, RecorderError>
where
    T: crate::ipc::bytewise::BytewiseReadOwned,
{
    // Verify CRC32 checksum
    let actual_crc32 = crc32fast::hash(data);
    if actual_crc32 != expected_crc32 {
        return Err(RecorderError::Corruption {
            details: format!(
                "CRC32 checksum mismatch: expected 0x{:08x}, got 0x{:08x}",
                expected_crc32, actual_crc32
            ),
        });
    }

    // Use new binary deserializer
    serializer::deserialize_record(data)
}

/// Create writer
fn create_writer<P: AsRef<Path>>(path: P) -> Result<ProtobufWriter, RecorderError> {
    ProtobufWriter::new(path)
}

/// Create reader
pub fn create_reader<P: AsRef<Path>>(path: P) -> Result<Box<dyn RecordReader>, RecorderError> {
    Ok(Box::new(ProtobufReader::new(path)?))
}

/// Create default header
fn create_default_header() -> RecorderHeader {
    RecorderHeader::new()
}

/// Helper function to write record block (using Protobuf serialization)
fn write_block(writer: &mut ProtobufWriter, block: &RecordBlock) -> Result<(), RecorderError> {
    writer.write_block(block)
}

/// Helper function to write file footer (using Protobuf serialization)
fn write_footer(writer: &mut ProtobufWriter, footer: &RecorderFooter) -> Result<(), RecorderError> {
    writer.write_footer(footer)
}

/// New request logger (simplified design)
pub struct RequestLogger {
    config: RecorderConfig,
    file_writer: ProtobufWriter,
    block_counter: AtomicU64,
    pending_blocks: VecDeque<RecordBlock>,
    last_flush: Instant,
    stats: RecorderStats,
    file_header: RecorderHeader,
}

impl RequestLogger {
    /// Create new logger
    pub fn new(config: RecorderConfig) -> Result<Self, RecorderError> {
        let mut file_writer = create_writer(&config.output_path)?;
        let file_header = create_default_header();

        // Write file header
        file_writer.write_header(&file_header)?;

        Ok(Self {
            config,
            file_writer,
            block_counter: AtomicU64::new(0),
            pending_blocks: VecDeque::new(),
            last_flush: Instant::now(),
            stats: RecorderStats::default(),
            file_header,
        })
    }

    /// Log a single writable object (using direct serialization)
    pub fn log_bytewise<T: BytewiseWrite>(&mut self, obj: &T) -> Result<(), RecorderError> {
        let block = self.create_record_block(obj)?;
        self.pending_blocks.push_back(block);

        // Force flush after batch write
        self.flush()
    }

    pub fn log_request<T: BytewiseWrite>(&mut self, obj: &T) -> Result<(), RecorderError> {
        self.log_bytewise(obj)
    }

    /// Batch log writable objects
    pub fn log_bytewises<T: BytewiseWrite>(&mut self, objs: &[T]) -> Result<(), RecorderError> {
        for obj in objs {
            let block = self.create_record_block(obj)?;
            self.pending_blocks.push_back(block);
        }

        // Force flush after batch write
        self.flush()
    }

    pub fn log_requests<T: BytewiseWrite>(&mut self, objs: &[T]) -> Result<(), RecorderError> {
        self.log_bytewises(objs)
    }

    /// Log a single serialized request
    pub fn log_request_serialized(&mut self, data: Vec<u8>) -> Result<(), RecorderError> {
        let block = self.create_record_block_serialized(data)?;
        self.pending_blocks.push_back(block);

        // Force flush after batch write
        self.flush()
    }

    /// Manually flush buffer
    pub fn flush(&mut self) -> Result<(), RecorderError> {
        for block in self.pending_blocks.drain(..) {
            write_block(&mut self.file_writer, &block)?;
            self.stats.total_requests += 1;
        }

        self.file_writer.flush()?;
        self.stats.flush_count += 1;
        self.stats.last_flush = Some(Instant::now());
        self.last_flush = Instant::now();

        Ok(())
    }

    /// Close logger and write file footer
    pub fn close(mut self) -> Result<(), RecorderError> {
        // Flush remaining data
        self.flush()?;

        // Write file footer
        let footer = RecorderFooter {
            blocks: self.stats.total_requests,
            sha256_digest: vec![], // TODO: Calculate actual SHA256
        };

        write_footer(&mut self.file_writer, &footer)?;
        self.file_writer.sync()?;

        Ok(())
    }

    /// Get recording statistics
    pub fn stats(&self) -> &RecorderStats {
        &self.stats
    }

    /// Get file header information
    pub fn header(&self) -> &RecorderHeader {
        &self.file_header
    }

    /// Create record block (using direct object serialization)
    fn create_record_block<T: BytewiseWrite>(&self, obj: &T) -> Result<RecordBlock, RecorderError> {
        let block_id = self.block_counter.fetch_add(1, Ordering::SeqCst);
        let timestamp_ns = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos() as u64;

        let thread_id = get_current_thread_id();

        // Use new direct serialization method
        let (obj_data, crc32_checksum) = serialize_bytewise(obj)?;

        let payload = RecordPayload {
            length: obj_data.len() as u64,
            crc32_checksum,
            request_data: obj_data,
        };

        Ok(RecordBlock {
            block_id,
            timestamp_ns,
            thread_id,
            payload: Some(payload),
        })
    }

    /// Create record block from serialized data
    fn create_record_block_serialized(&self, data: Vec<u8>) -> Result<RecordBlock, RecorderError> {
        let block_id = self.block_counter.fetch_add(1, Ordering::SeqCst);
        let timestamp_ns = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos() as u64;

        let thread_id = get_current_thread_id();

        let crc32_checksum = crc32fast::hash(&data);

        let payload = RecordPayload {
            length: data.len() as u64,
            crc32_checksum,
            request_data: data,
        };

        Ok(RecordBlock {
            block_id,
            timestamp_ns,
            thread_id,
            payload: Some(payload),
        })
    }

    /// Check if flush is needed
    fn should_flush(&self) -> bool {
        // Check buffer size
        if self.pending_blocks.len() >= self.config.buffer_size / 100 {
            return true;
        }

        // Check time interval (1 second)
        if self.last_flush.elapsed().as_secs() >= 1 {
            return true;
        }

        false
    }
}

impl Drop for RequestLogger {
    fn drop(&mut self) {
        // Try to flush remaining data, ignore errors
        let _ = self.flush();
    }
}

/// Extension trait: Add recording functionality to BytewiseWrite objects
pub trait BytewiseLogging {
    /// Record current object
    fn record_to(&self, logger: &mut RequestLogger) -> Result<(), RecorderError>;
}

impl<T: BytewiseWrite> BytewiseLogging for T {
    fn record_to(&self, logger: &mut RequestLogger) -> Result<(), RecorderError> {
        logger.log_bytewise(self)
    }
}
